package com.audaque.springboot.foshanupload.web.rocketmqdemo.controller;

import com.audaque.springboot.foshanupload.web.rocketmqdemo.enums.MessageType;
import com.audaque.springboot.foshanupload.web.rocketmqdemo.enums.SendType;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.message.MessageBuilder;
import org.apache.rocketmq.client.apis.producer.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.io.IOException;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;

@Slf4j
@RestController
@RequestMapping("clientProducer")
public class ClientProducerController {

    @Autowired
    private ClientConfiguration clientConfiguration;

    @GetMapping("send")
    public void send(
            @RequestParam(value = "topic") String topic, // 消息发送的目标Topic名称，需要提前创建。
            @RequestParam(value = "messageKey", required = false) String messageKey, //查询索引
            @RequestParam(value = "messagebBody") String messagebBody, //消息体
            @RequestParam(value = "messageTag", required = false) String messageTag,//过滤表达式
            @RequestParam(value = "sendTypeValue", required = false) Integer sendTypeValue,//发送方式
            @RequestParam(value = "delayMs", required = false) Long delayMs, //延时毫秒数:延时消息
            @RequestParam(value = "messageGroup", required = false) String messageGroup,//消息组：顺序消息
            @RequestParam(value = "messageTypeValue", required = false) Integer messageTypeValue,//消息类型
            @RequestParam(value = "businessKey", required = false) String businessKey//业务key：事务消息必须

    ) throws ClientException, IOException {
        // 普通消息发送
        ClientServiceProvider provider = ClientServiceProvider.loadService();

        MessageBuilder messageBuilder = provider.newMessageBuilder();
        //必要参数
        messageBuilder.setTopic(topic).setBody(messagebBody.getBytes());
        //查询索引
        if (StringUtils.isNotBlank(messageKey)) {
            // 设置消息索引键，可根据关键字精确查找某条消息
            messageBuilder.setKeys(messageKey);
        }
        // 设置消息Tag，用于消费端根据指定Tag过滤消息
        if (StringUtils.isNotBlank(messageTag)) {
            messageBuilder.setTag(messageTag);
        }
        //延时
        if (delayMs != null && delayMs > 0L) {
            messageBuilder.setDeliveryTimestamp(System.currentTimeMillis() + delayMs);
        } else if (StringUtils.isNotBlank(messageGroup)) {
            // 设置顺序消息的排序分组，该分组尽量保持离散，避免热点排序分组
            messageBuilder.setMessageGroup(messageGroup);
        } else if (StringUtils.isNotBlank(businessKey)) {
            //业务key
            // 一般事务消息都会设置一个本地事务关联的唯一ID，用来做本地事务回查的校验
            messageBuilder.addProperty(businessKey, UUID.randomUUID().toString());
        }
        Message message = messageBuilder.build();


        //必要参数
        ProducerBuilder producerBuilder = provider.newProducerBuilder()
                .setClientConfiguration(clientConfiguration)
                .setTopics(topic);
        Producer producer;
        Transaction transaction = null;
        if (Objects.equals(messageTypeValue, MessageType.TRANSACTION.getValue())) {
            // 设置事务检查器
            producerBuilder.setTransactionChecker(messageView -> {
                // 事务检查器一般是根据业务的ID去检查本地事务是否正确提交还是回滚，此处以订单ID属性为例
                // 在订单表找到了这个订单，说明本地事务插入订单的操作已经正确提交；如果订单表没有订单，说明本地事务已经回滚
                String orderId = messageView.getProperties().get(businessKey);
                if (StringUtils.isBlank(orderId)) {
                    // 没有业务ID直接回滚
                    return TransactionResolution.ROLLBACK;
                }
                // 检查本地事务是否提交
                String order = getOrderById(orderId);
                log.debug("check transaction start order={} [orderId={}]", order, orderId);
                if (StringUtils.isBlank(order)) {
                    // 本地事务没有正常提交直接回滚
                    return TransactionResolution.ROLLBACK;
                }
                // 通过业务ID查询到了对应的记录说明本地事务已经正常提交了
                // 消息事务也正常提交
                return TransactionResolution.COMMIT;
            });
            producer = producerBuilder.build();

            // 开启事务分支
            try {
                transaction = producer.beginTransaction();
            } catch (ClientException e) {
                e.printStackTrace();
                // 事务分支开启失败则直接退出
                return;
            }
        } else {
            producer = producerBuilder.build();
        }


        switch (SendType.get(sendTypeValue)) {
            case SYNC:
                SendReceipt sendReceipt;
                if (Objects.equals(messageTypeValue, MessageType.TRANSACTION.getValue())) {
                    // 发送事务消息
                    try {
                        sendReceipt = producer.send(message, transaction);
                    } catch (ClientException e) {
                        e.printStackTrace();
                        // 事务消息发送失败，事务可以直接退出并回滚
                        return;
                    }


                    /**
                     * 执行本地事务，并确定本地事务结果
                     * 1. 如果本地事务提交成功，则提交消息事务
                     * 2. 如果本地事务提交失败，则回滚消息事务
                     * 3. 如果本地事务未知异常，则不处理，等待事务消息回查
                     */
                    boolean localTransactionOk = doLocalTransaction();
                    if (localTransactionOk) {
                        try {
                            transaction.commit();
                        } catch (ClientException e) {
                            // 业务可以自身对实时性的要求选择是否重试，如果放弃重试，可以依赖事务消息回查机制进行事务状态的提交
                            e.printStackTrace();
                        }
                    } else {
                        try {
                            transaction.rollback();
                        } catch (ClientException e) {
                            // 建议记录异常信息，回滚异常时可以无需重试，依赖事务消息回查机制进行事务状态的提交
                            e.printStackTrace();
                        }
                    }
                } else {
                    // 同步消息发送
                    try {
                        // 发送消息，需要关注发送结果，并捕获失败等异常。
                        sendReceipt = producer.send(message);
                        log.debug("send message successfully, messageId={}", sendReceipt.getMessageId());
                    } catch (ClientException e) {
                        log.error("failed to send message", e);
                    }

                }
                break;
            case ASYNC:
                // 异步消息发送
                CompletableFuture<SendReceipt> completableFuture = producer.sendAsync(message); // 返回一个task编排工具
                // 回调处理
                completableFuture.whenComplete((receipt, exception) -> {
                    if (null != exception) {
                        exception.printStackTrace();
                    } else {
                        log.debug("send message successfully, messageId={}", receipt.getMessageId());
                        try {
                            producer.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                });
                break;
        }


        // 关闭
        //producer.close();
    }

    // 模拟订单查询服务用来确认订单事务是否提交成功
    private static String getOrderById(String orderId) {
        return "order";
    }

    // 模拟本地事务执行结果
    private static boolean doLocalTransaction() {
        return true;
    }


}
